package com.song.shop.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author Song Yinghui
 * @date 2018/8/24
 * @Version 1.0
 */
public class Customer {
    public static void main(String[] args) throws MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumerGroup");

        consumer.setNamesrvAddr("192.168.232.136:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("songtopic","*");
        consumer.registerMessageListener((MessageListenerConcurrently)(list , context)->{
            list.forEach(item ->{
                System.out.println("1111----"+ new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()));
                System.out.println(new String(item.getBody()));
            });
            System.out.println(list.size());
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        /*consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) -> {
            consumeOrderlyContext.setAutoCommit(true);
            list.forEach( o -> {
                System.out.println("-----------");
                System.out.println(new String(o.getBody()));
            });
            return ConsumeOrderlyStatus.SUCCESS;
        });*/

        consumer.start();
    }
}
